#include <Comm/Include/TCPSocket.h>

TCPSocket::TCPSocket(SOCKET _s, SyncQueue* TxQueue, SyncQueue* RxQueue, Semaphore* _sem )
{
    this->s = _s;
    this->sem_sync_owner = _sem;
    this->TxQueue = TxQueue;
    this->RxQueue = RxQueue;
    //this->sending_monitor = new Monitor();
    // parametros de seguridad para los threads lanzados
    SECURITY_ATTRIBUTES thread_security_att;
    thread_security_att.nLength = sizeof(SECURITY_ATTRIBUTES);
    thread_security_att.lpSecurityDescriptor = NULL;
    thread_security_att.bInheritHandle = TRUE;
    // inicializo el sender
    IPC_Thread_Data* sender_data = new IPC_Thread_Data[1];
    sender_data->ipc = this;
    sender_data->queue = TxQueue;
    sender_data->sem = this->sem_sync_owner;
    this->sender_thread = AfxBeginThread(SenderThreadFunction, (void*)sender_data, THREAD_PRIORITY_NORMAL, 0, CREATE_SUSPENDED, &thread_security_att);
    // inicializo el sender
    IPC_Thread_Data* receiver_data = new IPC_Thread_Data[1];
    receiver_data->ipc = this;
    receiver_data->queue = RxQueue;
    receiver_data->sem = this->sem_sync_owner;
    this->receiver_thread = AfxBeginThread(ReceiverThreadFunction, (void*)receiver_data, THREAD_PRIORITY_NORMAL, 0, CREATE_SUSPENDED, &thread_security_att);
    IniciarThreads();
}

TCPSocket::~TCPSocket(void)
{    
}

void TCPSocket::terminate()
{
    terminateReading();
    terminateWriting();
    closesocket(this->s);
}

SimpleMessage* TCPSocket::read()
{
    char* data = new char[TCP_MSG_SIZE];
    char* msg_data;
    unsigned int msg_size = 0;
    unsigned int tcp_msg_max_util_size = TCP_MSG_SIZE - sizeof(unsigned int);
    bool hay_siguiente_parte = true;
    int parte_nro = 0;
    int leido = 0;

    while (hay_siguiente_parte)
    {
        int bytes_left = TCP_MSG_SIZE;
        while ( bytes_left > 0 )
        {
            int bytes_read = recv(this->s, &(data[TCP_MSG_SIZE - bytes_left]), bytes_left, 0);
            if (0 == bytes_read)
            {
                delete data;
                throw WRITING_CLOSED_BY_OTHER_END;
            }
            else
            {
                if (SOCKET_ERROR == bytes_read)
                {
                    int error = WSAGetLastError();
                    switch (error)
                    {
                        case WSAECONNRESET:
                        {
                            delete data;
                            throw WRITING_CLOSED_BY_OTHER_END;
                            break;
                        }
                        case WSAESHUTDOWN:
                        case WSAECONNABORTED:
                        case WSAEINTR:
                        {
                            // si intento recibir tras hacxer un shutdown con SD_RECEIVE
                            delete data;
                            throw READING_CLOSED;
                            break;
                        }
                        default:
                        {
                            throw -1;
                            break;
                        }
                    }
                }
            }
            bytes_left -= bytes_read;
        }

        unsigned int bytes_util = *((unsigned int*)data);
        if (0 == parte_nro)
        {
            msg_data = new char[bytes_util];
            msg_size = bytes_util;
        }

        if ( bytes_util >= tcp_msg_max_util_size )
        {
            leido = tcp_msg_max_util_size;
        }
        else
        {
            leido = msg_size % (tcp_msg_max_util_size);
        }

        memcpy(msg_data + parte_nro * tcp_msg_max_util_size, data + sizeof(unsigned int), leido );

        if (bytes_util > tcp_msg_max_util_size )
        {
            parte_nro++;
        }
        else
        {
            hay_siguiente_parte = false;
        }
    }
    
    SimpleMessage* msg = new SimpleMessage( msg_data, msg_size );

    delete msg_data;
    
    delete data;

    return msg;
}

void TCPSocket::write(SimpleMessage *msg)
{
    char* data = new char[TCP_MSG_SIZE];
    unsigned int tcp_msg_max_util_size = TCP_MSG_SIZE - sizeof(unsigned int);

    char* msg_data = 0;
    unsigned int msg_data_size = msg->getData(msg_data);
    delete msg;

    bool hay_siguiente_parte = true;
    int parte_nro = 0;
    int copiado = 0;
    while (hay_siguiente_parte)
    {
        memcpy(data, &msg_data_size, sizeof(unsigned int));
        if ( msg_data_size >= tcp_msg_max_util_size )
        {
            copiado = tcp_msg_max_util_size;
        }
        else
        {
            copiado = msg_data_size % (tcp_msg_max_util_size);
        }
        memcpy(data + sizeof(unsigned int), msg_data + parte_nro * tcp_msg_max_util_size, copiado );
        int bytes_left = TCP_MSG_SIZE;
        while (bytes_left > 0)
        {
            int bytes_send = send(this->s, &(data[TCP_MSG_SIZE-bytes_left]), bytes_left, 0);
            if (SOCKET_ERROR == bytes_send)
            {
                int error = WSAGetLastError();
                switch (error)
                {
                    case WSAENOTSOCK: // chequear este case, si corresponde. Ver si no hay q poner otra excepcion
                    case WSAECONNRESET:
                    case WSAECONNABORTED:
                    {
                        delete data;
                        delete msg_data;
                        throw READING_CLOSED_BY_OTHER_END;
                        break;
                    }
                    default:
                    {
                        throw -1;
                    }                        
                }
            }
            bytes_left -= bytes_send;
        }
        if ( msg_data_size > tcp_msg_max_util_size)
        {
            msg_data_size -= copiado;
            parte_nro++;
        }
        else
        {
            hay_siguiente_parte = false;
        }
    }
    delete msg_data;
    delete data;
}

void TCPSocket::terminateWriting()
{
	//this->sending_monitor->enter();
    shutdown(this->s, SD_SEND);
	//this->sending_monitor->exit();
}

void TCPSocket::terminateReading()
{
    shutdown(this->s, SD_RECEIVE);
}

void TCPSocket::IniciarThreads()
{
    this->sender_thread->ResumeThread();
    this->receiver_thread->ResumeThread();
}

unsigned int TCPSocket::ReceiverThreadFunction(void *params)
{
    IPC_Thread_Data* thread_data = (IPC_Thread_Data*) params;
    bool fin = false;
    while (!fin)
    {
        SimpleMessage* msg = NULL;
        try
        {
            msg = thread_data->ipc->read();
        }
        catch (IPCResource_Exceptions e)
        {
            /// \todo
            switch (e)
            {
            case WRITING_CLOSED_BY_OTHER_END:
                {
                    fin = true;
                    thread_data->queue->close();
                    break;
                }
            case READING_CLOSED:
                {
                    fin = true;
                    // la cola es cerrada por terminateReceiving del canal
                    break;
                }
            default:
                {
                    e = COMMUNICATION_LOST;
                    fin = true;
                    break;
                }
            }
        }
        if (!fin)
        {
            try
            {
                thread_data->queue->push(msg);
            }
            catch (SyncQueue::SyncQueueExceptionType exception)
            {
                fin = true; 
                // ya fue cerrado el ipc
                delete msg;                
            }
        }
    }
    thread_data->sem->signal();
    delete thread_data;
    return 0;
}

unsigned int TCPSocket::SenderThreadFunction(void *params)
{
    IPC_Thread_Data* thread_data = (IPC_Thread_Data*) params;
    bool fin = false;
    //thread_data->ipc->sending_monitor->enter();
    while (!fin)
    {
        SimpleMessage* msg = NULL;
        try
        {
            msg = thread_data->queue->pop();
        }
        catch (SyncQueue::SyncQueueExceptionType queue_closed_exception)
        {
            fin = true;            
        }

        if (!fin)
        {
            try
            {
                thread_data->ipc->write(msg);
            }
            catch (IPCResource_Exceptions e)
            {
                switch (e)
                {
                case READING_CLOSED_BY_OTHER_END:
                    {
                        fin = true;
                        thread_data->queue->closeAndDiscardAll();
                        break;
                    }
                default:
                    {
                        throw -1;
                    }
                }
            }
        }
    }
    //thread_data->ipc->sending_monitor->exit();
    thread_data->sem->signal();
    delete thread_data;
    return 0;
}
